package com.hmall.search.canal.listeners;

import co.elastic.clients.elasticsearch.ElasticsearchClient;
import co.elastic.clients.elasticsearch.core.BulkRequest;
import co.elastic.clients.elasticsearch.core.BulkResponse;
import co.elastic.clients.elasticsearch.core.DeleteByQueryResponse;
import com.hmall.search.domain.po.ItemSync;
import org.springframework.amqp.core.ExchangeTypes;
import org.springframework.amqp.core.Message;
import org.springframework.amqp.rabbit.annotation.*;
import org.springframework.stereotype.Component;

import javax.annotation.Resource;
import java.io.IOException;
import java.util.List;
import java.util.stream.Collectors;

/**
 * @author Mr.M
 * @version 1.0
 * @description 商品信息同步程序
 * @date 2024/10/15 9:47
 */
@Component
public class ItemCanalDataSyncHandler extends AbstractCanalRabbitMqMsgListener<ItemSync> {

    @Resource
    private ElasticsearchClient esClient;

    //监听mq接收canal发送的信息，监听网关为exchange.canal-hmall,与此网关绑定的队列是canal-mq-hmall-item
    @RabbitListener(bindings = @QueueBinding(
            exchange = @Exchange(value = "exchange.canal-hmall", type = ExchangeTypes.TOPIC),
            value = @Queue(value = "canal-mq-hmall-item",arguments={@Argument(name="x-single-active-consumer", value = "true", type = "java.lang.Boolean")}),
            key = "canal-mq-hmall-item"
    ), concurrency="1")
    public void onMessage(Message message) throws Exception{
        parseMsg(message);
    }


    @Override
    public void batchSave(List<ItemSync> data) {
        BulkRequest.Builder br = new BulkRequest.Builder();
        for (ItemSync itemSync : data) {
            String[] strings = {itemSync.getCategory(), itemSync.getBrand()};
            itemSync.setSuggestion(strings);
            br.operations(op -> op
                    .index(idx -> idx
                            .index("items")
                            .id(itemSync.getId().toString())
                            .document(itemSync)
                    )
            );
        }

        BulkResponse result = null;
        try {
            result = esClient.bulk(br.build());
        } catch (IOException e) {
            throw new RuntimeException(e);
        }
        Boolean aBoolean = result.errors();
        if(aBoolean) {
            try {
                Thread.sleep(1000);
            } catch (InterruptedException e) {
                throw new RuntimeException(e);
            }
            throw new RuntimeException("同步失败");
        }
    }

    @Override
    public void batchDelete(List<Long> ids) {
        List<String> idList = ids.stream().map(id -> id.toString()).collect(Collectors.toList());
        DeleteByQueryResponse response = null;
        try {
            response = esClient.deleteByQuery(dq -> dq
                    .query(t -> t.ids(t1 -> t1.values(idList)))
                    .index("items"));
            boolean hasFailures = response.failures().size() > 0;
            if(hasFailures) {
                try {
                    Thread.sleep(1000);
                } catch (InterruptedException e) {
                    throw new RuntimeException(e);
                }
                throw new RuntimeException("同步失败");
            }
        } catch (IOException e) {
            throw new RuntimeException("同步失败");
        }
    }
}
